#!/usr/bin/env node
import { connect } from 'amqplib';
import { configure, getLogger } from 'log4js';
import { createClient, CreateMode, State } from 'node-zookeeper-client';
import uuid from 'uuid';
import config from './config';
import { IHeaders, IMessageContent } from './interfaces';
import send_msg from './send-msg';

const logger = getLogger();

const ZK_PATH = '/service_sc';
const JS_ROUTER_EXCHANGE = 'ex_service_router';

async function init_zookeeper() {
	const zk = createClient(config.ZK_HOSTS, config.ZK_OPTIONS || {});
	zk.connect();
	await (() => {
		return new Promise<void>((res) => {
			zk.on('state', (state) => {
				if (state === State.SYNC_CONNECTED) {
					logger.warn(`zookeeper connected, sessionid=${zk.getSessionId().toString('hex')}`, __filename);
					res();
				} else {
					logger.warn(`zookeeper state:${state}, sessionid=${zk.getSessionId().toString('hex')}`, __filename);
					try {
						zk.close();
					} catch (e) {
						logger.warn('zookeeper could not be closed.', __filename);
					}
					process.exit(-5);
				}
			});
		});
	})();
	try {
		// create path
		await (() => {
			return new Promise<void>((resolve, reject) => {
				zk.mkdirp(ZK_PATH, (error, _path) => {
					if (error) {
						logger.error(`Zookeeper:${error.toString()}. Path:${_path}. sessionid=${zk.getSessionId().toString('hex')}`, __filename);
						reject(error);
					} else {
						logger.warn(`Node: ${ZK_PATH} is created, path=${_path}. sessionid=${zk.getSessionId().toString('hex')}`, __filename);
						resolve();
					}
				});
			});
		})();
		// Then create node
		zk.create(
			`${ZK_PATH}/${config.SOA_ID}::${config.MESSAGE_TYPE}::service-${uuid()}`,
			new Buffer(''),
			CreateMode.EPHEMERAL,
			(error, _path) => {
				if (error) {
					logger.warn(`Node ${_path} could not be created. detail:${error.toString()}, sessionid=${zk.getSessionId().toString('hex')}`, __filename);
				} else {
					logger.warn(`Node: ${_path} is created. sessionid=${zk.getSessionId().toString('hex')}`, __filename);
					// logger.warn(`Node: %s is created. sessionid=${zk.getSessionId().toString('hex')}`, _path);
				}
			}
		);
		return zk;
	} catch (err) {
		logger.error(err.message, __filename);
		process.exit(-1);
	}
}

const APPID = 'nodejs-server';

async function init_rabbit() {
	logger.warn('Connecting to Rabbitmq server...........^v^', __filename);
	try {
		const conn = await connect(config.RABBITMQ);

		conn.on('error', (err: Error) => {
			logger.error(err.message, __filename);
			logger.warn('Node.js service error...........^v^', __filename);
			process.exit(-2);
		});
		conn.on('close', () => {
			logger.warn('Node.js service stoped...........^v^', __filename);
			process.exit(-3);
		});
		const ch = await conn.createChannel();
		ch.prefetch(config.PREFETCH_COUNT || 0);
		const q = await ch.assertQueue(config.JS_QUEUE_NAME, {
			autoDelete: false,
			durable: true,
			exclusive: false
		});
		const queue_name = q.queue;
		await ch.assertExchange(JS_ROUTER_EXCHANGE, 'direct', {
			autoDelete: false,
			durable: true
		});
		await ch.bindQueue(queue_name, JS_ROUTER_EXCHANGE, config.MESSAGE_TYPE);
		ch.consume(queue_name, async (msg) => {
			const props = msg.properties;
			const headers = props.headers as IHeaders;
			const { actionid, sessionid } = headers;
			const tm = new Date();
			try {
				const body = msg.content && msg.content.toString();
				const dbgmsg = 'Body=' + body + 'Props=' + JSON.stringify(props);
				logger.info('Message incomming:' + dbgmsg, __filename, (props as unknown as { message_id: string; }).message_id, actionid);
				const content = JSON.parse(body) as IMessageContent;
				const result = await send_msg(actionid, sessionid, 'dataservice.nodejs', content);
				const data = JSON.stringify(result);
				logger.info('Result:' + (data.length > 200 ? data.substr(0, 200) + '...' : data), __filename);
				ch.sendToQueue(props.replyTo as string, Buffer.from(data), {
					appId: APPID,
					correlationId: props.correlationId,
					deliveryMode: config.DELIVERY_MODE,
					headers: {
						actionid,
						'messagecontrol.code': 1,
						'messagecontrol.code_msg': '',
						sessionid
					}
				});
			} catch (err) {
				logger.trace(err);
				const e_msg = (err && err.message) || (err && err.toString && err.toString()) || '';
				logger.error('Service Error:' + e_msg, __filename, '', actionid);
				const err_msg = e_msg.length > 255 ? e_msg.substring(0, 100) + '...' : e_msg;
				ch.sendToQueue(props.replyTo as string, Buffer.from(e_msg), {
					appId: APPID,
					correlationId: props.correlationId,
					deliveryMode: config.DELIVERY_MODE,
					headers: {
						actionid,
						'messagecontrol.code': 0,
						'messagecontrol.code_msg': err_msg,
						sessionid
					}
				});
			} finally {
				ch.ack(msg);
				const cost = new Date().getTime() - tm.getTime();
				const dbgmsg = 'Props=' + JSON.stringify(props);
				logger.info('Message outgoing:' + dbgmsg + 'Costing', __filename, (props as unknown as { message_id: string; }).message_id, actionid, cost);
				if (cost > 500) {
					const body = msg.content && msg.content.toString();
					logger.error(`Service cost ${cost} ms, Message body=[${body}]`, __filename, (props as unknown as { message_id: string; }).message_id, actionid, cost);
				} else if (cost > 200) {
					const body = msg.content && msg.content.toString();
					logger.warn(`Service cost ${cost} ms, Message body=[${body}]`, __filename, (props as unknown as { message_id: string; }).message_id, actionid, cost);
				}
			}
		});
		logger.warn('Connected to Rabbitmq server...........^v^', __filename);
	} catch (err) {
		logger.trace(err);
		logger.warn('Node.js service is crashed...........^v^', __filename);
		logger.error(err.message, __filename);
		process.exit(-4);
	}
}

async function main() {
	process.on('SIGINT', () => {
		process.exit(0);
	});

	configure('./log4js.json');
	logger.warn('Starting Node.js service...........^v^', __filename);
	await init_zookeeper();
	await init_rabbit();
	logger.warn('Node.js service is started...........^v^', __filename);
}

main();
